package com.didi.gulimall.listener;

import com.alibaba.fastjson.JSON;
import com.didi.gulimall.common.constant.CartConstant;
import com.didi.gulimall.common.constant.RabbitMqConstant;
import com.didi.gulimall.common.constant.SecKillConstant;
import com.didi.gulimall.common.entity.LocalMqEntity;
import com.didi.gulimall.common.to.mq.SeckillOrderTo;
import com.didi.gulimall.common.utils.LogUtils;
import com.didi.gulimall.seckill.service.SecKillService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * @Author 10626
 * @Create 2021/6/4 11:30
 */
@Slf4j
@Component
public class SecKillConsumerMqListener {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private SecKillService secKillService;

    @RabbitListener(queues = "seckill.release.order.queue")
    public void receiveMessageOrder(SeckillOrderTo seckillOrderTo, Message message, Channel channel) throws IOException {
        MessageProperties messageProperties = message.getMessageProperties();
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        String messageId = (String) headers.get("spring_returned_message_correlation");
        long deliveryTag = messageProperties.getDeliveryTag();
        String cartKey = SecKillConstant.SKUKILL_MQ_PREFIX + messageId;

        log.info("秒杀收到消息key=" + cartKey + JSON.toJSONString(seckillOrderTo));

        String json = redisTemplate.opsForValue().get(cartKey);
        LocalMqEntity localMqEntity = JSON.parseObject(json, LocalMqEntity.class);

        if (localMqEntity == null || localMqEntity.getStatus().equals(RabbitMqConstant.CONSUMED)) {
            //   该订单已经回滚
            channel.basicAck(deliveryTag, false);
            if (localMqEntity != null) {
                localMqEntity.setStatus(RabbitMqConstant.CONSUMED);
                redisTemplate.opsForValue().set(cartKey, JSON.toJSONString(localMqEntity));
            }
        } else {
            // 重试10次就不要再发了
            if (localMqEntity.getRetries() < RabbitMqConstant.MAXIMUM_NUMBER_OF_RETRIES) {
                try {
                    secKillService.rollbackOrder(seckillOrderTo);
                    channel.basicAck(deliveryTag, false);
                    localMqEntity.setStatus(RabbitMqConstant.CONSUMED);
                    redisTemplate.opsForValue().set(cartKey, JSON.toJSONString(localMqEntity));
                } catch (Exception e) {
                    log.error(LogUtils.getStackTraceInfo(e));
                    localMqEntity.setRetries(localMqEntity.getRetries() + 1);
                    redisTemplate.opsForValue().set(cartKey, JSON.toJSONString(localMqEntity));
                    channel.basicReject(deliveryTag, true);
                }
            } else {
                // 消息丢弃
                channel.basicReject(deliveryTag, false);
            }
        }
    }
}